package com.win.service.impl;

import com.win.common.HdfsCommon;
import com.win.mapreduce.WordCountDriver;
import com.win.pojo.vo.EchartsInfoVo;
import com.win.service.EchartsService;
import com.win.service.HdfsService;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
public class EchartsServiceImpl implements EchartsService {

    @Autowired
    private WordCountDriver wordCountDriver;
    @Autowired
    private HdfsService hdfsService;

    @Override
    public Map<String, List<EchartsInfoVo>> getEchartsInfo() {
        String path = "/mapreduce/output";

        Map<String, List<EchartsInfoVo>> map = new HashMap<>();

        try {
            //执行mapreduce job之前先删除目标文件夹
            deleteFolder(path);
            //运行job
            int res = wordCountDriver.executeMapReduceJob(path);
            //说明mapreduce job运行成功
            if (res == 0){
                FileSystem fs = FileSystem.get(new URI(HdfsCommon.hdfsUri), new Configuration());
                //获取mapreduce结果的文件名称
                List<String> fileNameList = hdfsService.getFileName(fs, path);

                List<EchartsInfoVo> echartsInfoVoList1 = null;
                List<EchartsInfoVo> echartsInfoVoList2 = null;
                //遍历文件集合，获取存放wordcount结果的文件
                for (String fileName : fileNameList) {
                    if (fileName.equals("part-r-00000")){
                        //此文件中存放获取到的交通密度的数据
                        echartsInfoVoList1 = getInfo(new Configuration(), path + "/" +fileName);
                    }
                    if (fileName.equals("part-r-00001")){
                        //此文件夹存放的是获取到的空气质量的数据
                        echartsInfoVoList2 = getInfo(new Configuration(), path + "/" + fileName);
                    }
                }
            map.put("trafficDensityInfo",echartsInfoVoList1);
            map.put("airQualityInfo",echartsInfoVoList2);
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(e);
        } catch (URISyntaxException e) {
            throw new RuntimeException(e);
        }
        return map;
    }

    private List<EchartsInfoVo> getInfo(Configuration configuration,String path) throws URISyntaxException, IOException {

        FileSystem fs = FileSystem.get(new URI(HdfsCommon.hdfsUri), configuration);

        List<EchartsInfoVo> echartsInfoVoList = new ArrayList<>();

        FSDataInputStream inputStream = fs.open(new Path(path));
        BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));

        String line;
        while ((line = reader.readLine()) != null){
            String[] infos = line.split("\\s+");

            EchartsInfoVo echartsInfoVo = EchartsInfoVo.builder()
                    .name(infos[0])
                    .value(infos[1])
                    .build();
            echartsInfoVoList.add(echartsInfoVo);
            //System.out.println(echartsInfoVo);
        }
        return echartsInfoVoList;
    }

    private void deleteFolder(String path) throws IOException {
        Configuration conf = new Configuration();
        // 如果需要，可以添加HDFS的配置，例如HDFS的NameNode地址
        // conf.set("fs.defaultFS", hdfsUri);

        FileSystem fs = FileSystem.get(URI.create(HdfsCommon.hdfsUri), conf);

        // 检查文件夹是否存在
        if (fs.exists(new Path(path))) {
            // 递归删除文件夹及其内容
            boolean success = fs.delete(new Path(path), true);
            if (success) {
                System.out.println("Folder " + path + " deleted successfully.");
            } else {
                System.out.println("Failed to delete folder " + path);
            }
        }
        // 关闭文件系统连接
        fs.close();
    }

}
